package com.idanchuang.count.listener

import com.google.common.eventbus.EventBus
import com.idanchuang.count.config.IntegrationTestConfig
import com.idanchuang.count.core.constant.CountConstants
import com.idanchuang.count.core.entity.SimpleCountIncrEvent
import com.idanchuang.count.core.model.BizProperties
import com.idanchuang.count.dao.CountDao
import com.idanchuang.count.service.SimpleCountService
import com.idanchuang.count.service.api.request.query.CountBatchQuery
import com.idanchuang.count.service.api.request.query.CountQuery
import com.idanchuang.count.service.impl.SimpleCountServiceImpl
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.data.redis.core.StringRedisTemplate
import org.springframework.kafka.support.Acknowledgment
import org.springframework.test.context.ContextConfiguration
import org.testcontainers.spock.Testcontainers
import spock.lang.Specification
import spock.lang.Subject

import static com.idanchuang.count.core.util.keyformat.SimpleCountKeysFormatter.formatCountKey
import static com.idanchuang.count.core.util.keyformat.SimpleCountKeysFormatter.formatUidKey

/**
 * @author Richard_yyf
 * @version 1.0 2021/9/26
 */
@ContextConfiguration(
        initializers = IntegrationTestConfig.IntegrationTestInitializer.class,
        classes = [IntegrationTestConfig.class])
@Testcontainers
@Subject([SimpleCountIncrEventListener, SimpleCountServiceImpl])
class SimpleCountIncrEventListenerTest extends Specification {

    @Autowired
    private BizProperties bizProperties

    @Autowired
    private CountDao countDao

    @Autowired
    private StringRedisTemplate stringRedisTemplate


    @Autowired
    private SimpleCountService simpleCountService

    private EventBus kafkaSendEventBus = Mock()

    private Long countCommandIdempotentExpire = 86400L

    private SimpleCountIncrEventListener listener

    private Acknowledgment ack = Mock()

    private String bizCodeA = "bizA"
    private Integer bizCodeAId = 1
    private String bizCodeBNeedNotify = "bizB"
    private Integer bizCodeBId = 2

    def setup() {
        listener = new SimpleCountIncrEventListener()
        listener.bizProperties = bizProperties
        listener.countDao = countDao
        listener.kafkaSendEventBus = kafkaSendEventBus
        listener.countCommandIdempotentExpire = countCommandIdempotentExpire
    }

    def "SingleConsumeIncrEvent_bizCode_normal"() {
        // 没有被幂等，正常流程
        given:
        def event = new SimpleCountIncrEvent(
                bizCode: bizCodeBNeedNotify,
                bizNo: System.currentTimeMillis(),
                increment: 1,
                time: String.valueOf(System.currentTimeMillis()/1000),
        )
        def uidKey = formatUidKey(bizCodeBId, event.bizNo, event.time)
        def countKey = formatCountKey(bizCodeBId, event.bizNo)
        when:
        listener.singleConsumeIncrEvent(event, ack)
        // 重复消费
        listener.singleConsumeIncrEvent(event, ack)

        def uidVal = stringRedisTemplate.opsForValue().get(uidKey)
        def val = countDao.getCount(countKey)

        def count = simpleCountService.query(new CountQuery(bizCode: event.bizCode, bizNo: event.bizNo))

        then:
        2 * ack.acknowledge()
        1 * kafkaSendEventBus.post(_)
        val == event.increment
        count == val
        // 幂等成功
        uidVal == CountConstants.UID_SET_VALUE
    }


    def "SingleConsumeIncrEvent_null"() {
        when:
        listener.singleConsumeIncrEvent(null, ack)
        then:
        notThrown()
    }

    def "SingleConsumeIncrEvent_bizCode_not_exist"() {
        given:
        SimpleCountIncrEvent event = new SimpleCountIncrEvent(
                bizCode: "notExist",
                bizNo: System.currentTimeMillis(),
                increment: 1,
                time: String.valueOf(System.currentTimeMillis()/1000)
        )

        when:
        listener.singleConsumeIncrEvent(event, ack)

        then:
        notThrown()
    }

    def "BatchConsumeIncrEvent"() {
        given:
        def eventA1 =  new SimpleCountIncrEvent(
                bizCode: bizCodeA,
                bizNo: "bizANo1" + System.currentTimeMillis(),
                increment: 1,
                time: String.valueOf(System.currentTimeMillis()/1000),
        )
        def eventA2 =  new SimpleCountIncrEvent(
                bizCode: bizCodeA,
                bizNo: "bizANo2" + System.currentTimeMillis(),
                increment: 2,
                time: String.valueOf(System.currentTimeMillis()/1000)
        )

        def eventB1 =  new SimpleCountIncrEvent(
                bizCode: bizCodeBNeedNotify,
                bizNo: "bizBNo1" + System.currentTimeMillis(),
                increment: 1,
                time: String.valueOf(System.currentTimeMillis()/1000)
        )
        def eventB2 =  new SimpleCountIncrEvent(
                bizCode: bizCodeBNeedNotify,
                bizNo: "bizBNo2" + System.currentTimeMillis(),
                increment: 2,
                time: String.valueOf(System.currentTimeMillis()/1000),
        )

        def countBatchQueryA = new CountBatchQuery(bizCode: bizCodeA, bizNos: [eventA1.bizNo, eventA2.bizNo])
        def countBatchQueryB = new CountBatchQuery(bizCode: bizCodeBNeedNotify, bizNos: [eventB1.bizNo, eventB2.bizNo])

        when:
        listener.batchConsumeIncrEvent([eventA1, eventA2, eventB1, eventB2], ack)
        def countMapA = simpleCountService.batchQuery(countBatchQueryA)
        def countMapB = simpleCountService.batchQuery(countBatchQueryB)

        then:
        1 * ack.acknowledge()
        // 两个值变更通知
        2 * kafkaSendEventBus.post(_)
        countMapA[eventA1.bizNo] == eventA1.increment
        countMapA[eventA2.bizNo] == eventA2.increment
        countMapB[eventB1.bizNo] == eventB1.increment
        countMapB[eventB2.bizNo] == eventB2.increment
    }
}
